package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/12/30 08:59
 */
public class Dwd_02_DwdCartAddApp extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_02_DwdCartAddApp().init(
            3002,
            2,
            "Dwd_02_DwdCartAddApp"
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db 数据
        //tEnv.executeSql("ddl 和 dml 中的: 增删改");
        //tEnv.sqlQuery("查询语句");
        readOdsDb(tEnv, "Dwd_02_DwdCartAddApp");
        
        
        // 2. 过滤出加购表数据
        Table cartInfo = tEnv.sqlQuery("select " +
                                           "`data`['id'] id," +
                                           "`data`['user_id'] user_id," +
                                           "`data`['sku_id'] sku_id," +
                                           "`data`['source_id'] source_id," +
                                           "`data`['source_type'] source_type," +
                                           "if(`type` = 'insert'," +
                                           "`data`['sku_num'], cast((cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int)) as string)) sku_num," +
                                           "ts," +
                                           "pt " +
                                           "from ods_db " +
                                           "where `database`='gmall2022' " +
                                           "and `table`='cart_info' " +
                                           "and `type`='insert' or (" +
                                           " `type`='update' " +
                                           "  and `old`['sku_num'] is not null " + // sku_num 发生了变化
                                           "  and cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int) " +
                                           ")");
        tEnv.createTemporaryView("cart_info", cartInfo);
        
        // 3. 读取维度表: 字典表
        readBaseDic(tEnv);
        // 4. 加购表和字典表 join
        Table result = tEnv.sqlQuery("select " +
                                         "ci.id," +
                                         "user_id," +
                                         "sku_id," +
                                         "source_id," +
                                         "source_type," +
                                         "dic_name source_type_name," +
                                         "sku_num," +
                                         "ts " +
                                         "from cart_info ci " +
                                         "join base_dic for system_time as of ci.pt as dic " +
                                         "on ci.source_type=dic.dic_code");
        // 5. 把 join 的结果写出到 dwd 层
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                            "id string," +
                            "user_id string," +
                            "sku_id string," +
                            "source_id string," +
                            "source_type_code string," +
                            "source_type_name string," +
                            "sku_num string," +
                            "ts bigint" +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));
    
        result.executeInsert("dwd_trade_cart_add");
    }
}
/*
交易域加购事务事实表

1. 数据源业务数据
	ods_db

	mysql: 关系型数据库(结构化数据)

	1. 使用 sql 技术更简单
	2. 在企业中, sql 其实更普遍

2. 过滤出加购表的数据
	cart_info

	insert
	update
		sku_num 变化,必须是变大

			sku_id    sku_num
	新增	  		1         3      ->3
	更新	        1         5      ->(5-3)=2

	delete
		不处理

3. 维度退化
	
	表的 join

4. 把结果写入到 dwd 层(Kafka)

----
总结:
    加购
    
    1. 封装 BaseSQLApp
    2. 通过 ddl 读取 kafka 的数据
    3. join
            常规
                join
                left join
                考虑 ttl
            间隔 join
            时态 join
                版本表
                
                基于处理时间
                基于事件时间
            lookup join
                基于处理时间 的时态 join 的特殊用法
                
                用来事实表与维度表的 join
     4. 加购表和字典表的 join
           lookup join
           
           lookup join 的不是去 phoenix 中 join?
             目前的 lookup join 中使用的jdbc 连接器不支持 phoenix
      5. 把 join 的结果写出到 dwd
            注意类型的匹配
 */